package com.dongnao.network.nio.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 主从Reactor，也叫多Reactor
 */
public class MultipleReactorServer {
    private ServerSocketChannel serverSocketChannel;
    private SubReactor[] subReactors = new SubReactor[16];//负责处理IO读写
    private MainReactor mainReactor;//负责接受客户端的连接
    private static ExecutorService workerPool = Executors.newCachedThreadPool();//处理业务操作的线程

    public static void main(String[] args) throws IOException {
        MultipleReactorServer server = new MultipleReactorServer();
        server.newGroup();
        server.initAndRegister();
        server.bind();
    }

    private void newGroup() throws IOException {
        mainReactor = new MainReactor();

        for (int i = 0; i < subReactors.length; i++) {
            subReactors[i] = new SubReactor();
        }
    }

    private void initAndRegister() throws IOException {
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        mainReactor.register(serverSocketChannel);

        mainReactor.start();
    }

    //绑定端口，启动服务
    private void bind() throws IOException {
        serverSocketChannel.bind(new InetSocketAddress(8080));
        System.out.println("----服务端启动----");
    }

    /**
     * 处理多个客户端并发请求
     * MainReactor只处理Accept IO事件
     */
    class MainReactor extends Thread {
        private Selector selector;
        AtomicInteger incr = new AtomicInteger(0);

        public MainReactor() throws IOException {
            selector = Selector.open();
        }

        public void register(ServerSocketChannel serverSocketChannel) {
            try {
                //获取serverSocket在Selector中注册的标识sk，sk绑定了serverSocket与Selector的关系，维护channel事件
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    selector.select();//阻塞，直到事件通知才会返回

                    Set<SelectionKey> selKeySet = selector.selectedKeys();
                    Iterator<SelectionKey> it = selKeySet.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        it.remove();
                        System.out.println("----MainReactor#run(), key isAcceptable: " + key.isAcceptable()
                                + ", isConnectable: " + key.isConnectable()
                                + ", isReadable: " + key.isReadable()
                                + ", isWritable: " + key.isWritable()
                                + ", isValid: " + key.isValid());

                        if (key.isAcceptable()) {
                            SocketChannel socketChannel = ((ServerSocketChannel) key.channel()).accept();
                            System.out.println("收到新的客户端连接，socketChannel: " + (socketChannel == null ? "null" : socketChannel));
                            //将客户端连接交给Acceptor
                            new Acceptor(socketChannel);
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        class Acceptor {
            public Acceptor(SocketChannel channel) {
                int index = incr.getAndIncrement();
                int idx = index % subReactors.length;
                subReactors[idx].register(channel);
                subReactors[idx].start();

                //避免溢出
                if (incr.get() == Integer.MAX_VALUE >> 1) {
                    incr.set(0);
                }
            }
        }
    }

    /**
     * 处理多个客户端并发请求
     */
    class SubReactor extends Thread {
        final Selector sel;
        SocketChannel socket;

        public SubReactor() throws IOException {
            sel = Selector.open();
        }

        public void register(SocketChannel socketChannel) {
            socket = socketChannel;
            try {
                socket.configureBlocking(false);
                SelectionKey key = socket.register(sel, SelectionKey.OP_READ);
                new Handler(key);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    sel.select();
                    Set<SelectionKey> selKeySet = sel.selectedKeys();
                    Iterator<SelectionKey> it = selKeySet.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        System.out.println("====SubReactor#run(), key isAcceptable: " + key.isAcceptable()
                                + ", isConnectable: " + key.isConnectable()
                                + ", isReadable: " + key.isReadable()
                                + ", isWritable: " + key.isWritable()
                                + ", isValid: " + key.isValid());
                        dispatch(key);
                    }
                    selKeySet.clear();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        private void dispatch(SelectionKey selectionKey) {
            Runnable r = (Runnable) selectionKey.attachment();
            if (r != null) {
                r.run();
            }
        }

        /**
         * 客户端读写事件处理器
         */
        class Handler implements Runnable {
            final SocketChannel socket;
            final SelectionKey sk;
            final int MAXIN = 1024;
            ByteBuffer input = ByteBuffer.allocate(MAXIN);
            ExecutorService threadPool = CustomThreadPool.buildPool();

            public Handler(SelectionKey key) throws IOException {
                System.out.println("----Handler()----");
                socket = (SocketChannel) key.channel();
                sk = key;
                sk.attach(this);//绑定自己为读写事件的处理器
                //selector.wakeup();//激活在select()方法上的线程，告诉它感兴趣的事件发生变化了
            }

            @Override
            public void run() {
                System.out.println("----Handler#run()----");
                try {
                    if (sk.isReadable()) {
                        read();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            void read() throws IOException {
                int len = socket.read(input);
                if (inputIsComplete()) {
                    input.flip();
                    //process();//读完数据，进行业务逻辑处理
                    threadPool.execute(new Processer(sk, input));
                    input.clear();
                }
                if (len < 0) {
                    sk.cancel();
                }
            }

            private boolean inputIsComplete() {
                return input.position() > 0;
            }
        }

        class Processer implements Runnable {
            final ByteBuffer input;
            final SelectionKey sk;

            public Processer(SelectionKey selKey, ByteBuffer in) {
                input = ByteBuffer.wrap(in.array(), 0, in.limit());
                sk = selKey;
            }

            @Override
            public void run() {
                //模拟业务逻辑处理
                System.out.println("----Processer#run()，收到数据：" + new String(input.array(), input.position(), input.limit()));
//            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(300));
                String msg = "Now time is : " + (new Date()).toString();
                ByteBuffer output = ByteBuffer.allocate(1024);
                output.put(msg.getBytes());
                try {
                    output.flip();
//                    new Sender(sk, output).run();
                      new Sender(sk, output);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

        class Sender implements Runnable {
            final SocketChannel socket;
            ByteBuffer output;
            SelectionKey sk;
            final Object oldAttachment;

            public Sender(SelectionKey selKey, ByteBuffer respMsg) {
                socket = (SocketChannel) selKey.channel();
                output = ByteBuffer.wrap(respMsg.array(), 0, respMsg.limit());
                sk = selKey;
                sk.interestOps(sk.interestOps() | SelectionKey.OP_WRITE);//readyOps与interestOps不一致
                oldAttachment = sk.attach(this);
                sk.selector().wakeup();
            }

            @Override
            public void run() {
                try {
                    System.out.println("----Sender#run()----");
                    if (sk.isWritable()) {
                        socket.write(output);
                        if (outputIsComplete()) {
                            output.clear();
                            sk.attach(oldAttachment);
                            //只有写事件操作完成了，才取消
                            sk.interestOps(sk.interestOps() & ~SelectionKey.OP_WRITE);
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            private boolean outputIsComplete() {
                return !output.hasRemaining();
            }
        }
    }
}
